package com.flink.test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * StreamWordCount
 *
 * @author caizhiyang
 * @since 2024-01-16
 */
public class StreamWordCount2 {

    /**
     * 流处理代码
     *
     * @param args
     */
    public static void main(String[] args) throws Exception {
        // 1. 创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2.读取文件数据
        DataStreamSource<String> lineStream = env.readTextFile("data/word.txt");

        // 3. 转换、分组、求和，得到统计结果
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            String[] words = line.split(" ");

            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG)).keyBy(data -> data.f0).sum(1);
        /*
        * 由于Java存在泛型擦除机制，因此有些时候使用lambda表达式时，无法获取泛型类型，因此需要显示提供类型
        *
        * returns(Types.TUPLE(Types.STRING, Types.LONG))
        * 这段代码就是显示指定返回类型
        * */


        // 4. 打印
        sum.print();

        // 5. 执行
        env.execute();
    }
}
